package com.shellyan.gmall.rt.app

import com.alibaba.fastjson.JSON
import com.atguigu.realtime.gmall.common.Constant
import com.shellyan.gmall.rt.bean.{OrderDetail, OrderInfo, SaleDetail}
import com.shellyan.gmall.rt.util.RedisUtil
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.dstream.DStream
import redis.clients.jedis.Jedis

/**
 * @author Shelly An
 * @create 2020/9/11 9:05
 *         需要两个流，因此BaseApp不能满足需求，提升一下，写一个Base2App
 */
object SaleDetailApp extends Base2App {
  override val master: String = "local[2]"
  override val appName: String = "SaleDetailApp"
  override val batchTime: Int = 3
  override val topics: Set[String] = Set(Constant.ORDER_INFO_TOPIC, Constant.ORDER_DETAIL_TOPIC)
  override val groupId: String = "SaleDetailApp"

  /**
   * 解决方式1
   *
   * @param orderInfoStream
   * @param orderDetailStream
   */
  def joinOrderInfoOrderDetail(orderInfoStream: DStream[OrderInfo], orderDetailStream: DStream[OrderDetail]): DStream[SaleDetail] = {
    val orderInfoStreamWithWindow: DStream[(String, OrderInfo)] = orderInfoStream
      .window(Seconds(batchTime * 3), Seconds(batchTime))
      .map((info: OrderInfo) =>(info.id,info))
    val orderDetailStreamWithWindow: DStream[(String, OrderDetail)] = orderDetailStream
      .window(Seconds(batchTime * 3), Seconds(batchTime))
      .map((detail: OrderDetail) =>(detail.order_id,detail))

    val saleDetailStream: DStream[SaleDetail] = orderInfoStreamWithWindow
      .join(orderDetailStreamWithWindow)
      .map{
        case(orderId,(orderInfo,orderDetail)) =>
          SaleDetail().mergeOrderInfo(orderInfo).mergeOrderDetail(orderDetail)
      }

    //saleDetailStream 内部会重复join，需要去重，使用redis去重
    saleDetailStream.mapPartitions((saleDetailIt: Iterator[SaleDetail]) => {
      val client: Jedis = RedisUtil.getClient
      //和日活的思路一样，把order_detail_id写入到redis的set中，如果返回1表示是第一次join（保留），如果返回0表示不是第一次（过滤掉）
      val result: Iterator[SaleDetail] = saleDetailIt.filter((saleDetail: SaleDetail) => {
        //一天一个key，过期的key会自动删除，否则只用一个key，这个set越来越大
        client.sadd(s"order_detail_ids:${saleDetail.dt}", saleDetail.order_detail_id) == 1
        //加一个过期时间
      })

      client.close()
      result
    })

  }

  override def run(streams: Map[String, DStream[String]]): Unit = {
    val orderInfoStream: DStream[OrderInfo] = streams(Constant.ORDER_INFO_TOPIC)
      .map(json => JSON.parseObject(json, classOf[OrderInfo]))
    val orderDetailStream: DStream[OrderDetail] = streams(Constant.ORDER_DETAIL_TOPIC)
      .map(json => JSON.parseObject(json, classOf[OrderDetail]))

    //1. 对两个流进行join内连接
    // join 内连接  leftJoin 左连接  rightJoin 右连接  fullJoin 全集
    val saleDetail: DStream[SaleDetail] = joinOrderInfoOrderDetail(orderInfoStream, orderDetailStream)
    saleDetail.print()
  }
}
